package net.bwie.realtime.vehicle.function;

import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.bwie.realtime.vehicle.bean.BatteryHealthData;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * 电池健康窗口函数
 * 该函数处理电池健康数据，计算每个窗口内电池充满电容量的平均值
 */
public class BatteryHealthWindowFunction implements WindowFunction<BatteryHealthData, String, String, TimeWindow> {
    @Override
    public void apply(String s, TimeWindow window, Iterable<BatteryHealthData> input, Collector<String> out) throws Exception {
        // 将窗口的开始和结束时间戳转换为可读的日期时间格式
        String startTime = DateTimeUtil.convertLongToString(window.getStart(), DateTimeUtil.DATE_TIME_FORMAT);
        String endTime = DateTimeUtil.convertLongToString(window.getEnd(), DateTimeUtil.DATE_TIME_FORMAT);
        // 初始化计数器和总和变量
        double count = 0;
        double sum = 0;
        // 遍历窗口内的所有电池健康数据
        for (BatteryHealthData element : input) {
            // 计数数据条数
            count++;
            // 累加所有电池的充满电容量
            sum += element.getFullChargeCapacity();
        }
        // 计算平均充满电容量，并与窗口时间、数据条数、总充满电容量和当前处理时间一起组成输出字符串
        String output = startTime + "," + endTime + ","
                + count + "," + sum + "," + (sum / count) + ","
                + System.currentTimeMillis();
        // 使用收集器输出处理结果
        out.collect(output);
    }
}
